package com.asdc.demo.confirm;

import com.asdc.demo.utils.RabbitMqUtils;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.ConfirmCallback;
import com.rabbitmq.client.MessageProperties;

import javax.swing.undo.CannotUndoException;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;

/**
 * @ClassName TaskConfirm01
 * @Author bsy
 * @Description 生产端消息可靠性 - 异步消息确认
 * <p>
 * 1、用Map改成异步
 * 2、支持高并发
 * 3、能很高的控制那条消息出现问题
 *
 * @Date 2022/3/12 16:18
 * @Version 1.0
 */
public class ProdConfirmAsyn01 {
    private static final String QUEUE_NAME = "QUEUE_CONFIR_VVYX";

    public static void main(String[] args) throws Exception {
        try (Channel channel = RabbitMqUtils.getChannel()) {
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            //开启消息确认机制
            channel.confirmSelect();
            ConcurrentSkipListMap<Object, Object> map = new ConcurrentSkipListMap<>();
            //确认消息
            ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
                if (multiple) {
                    //批量
                    ConcurrentNavigableMap<Object, Object> mulMap = map.headMap(deliveryTag);
                    mulMap.clear();
                } else {
                    map.remove(deliveryTag);
                }
            };

            //不确认消息
            ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
                String noAckMsg = (String) map.get(deliveryTag);
                System.out.println("=====NOAck: " + noAckMsg);
            };

            channel.addConfirmListener(ackCallback, nackCallback);
            long start = System.currentTimeMillis();
            for (int i = 0; i < 1000; i++) {
                String msg = "msg" + i;
                channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN, msg.getBytes());
                map.put(channel.getNextPublishSeqNo(), msg);
            }
            long end = System.currentTimeMillis();
            System.out.println("成功向队列发送消息-------用时: " + (end - start) + " ms");
        }
    }
}
